package com.etc.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.List;

/**
 * @author 杰哥
 * @Title: WindowHotWord
 * @ProjectName scalaworldcount
 * @Description: TODO
 * @date 2019/8/2118:49
 */
public class WindowHotWord {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("WindowHotWord").setMaster("local[*]");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
        JavaReceiverInputDStream<String> master = jsc.socketTextStream("master", 9999);
        JavaPairDStream<String, Integer> JavaPairDStream = master.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s.split(" ")[1],1);
            }
        });
        JavaPairDStream<String, Integer> IntegerJavaPairDStream = JavaPairDStream.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }, Durations.seconds(60), Durations.seconds(10));
        JavaPairDStream<String, Integer> stringIntegerJavaPairDStream = IntegerJavaPairDStream.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
            @Override
            public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> stringIntegerJavaPairRDD) throws Exception {
                JavaPairRDD<Integer, String> integerStringJavaPairRDD = stringIntegerJavaPairRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                    @Override
                    public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return new Tuple2<Integer, String>(stringIntegerTuple2._2,stringIntegerTuple2._1);

                    }
                });
                JavaPairRDD<Integer, String> integerStringJavaPairRDD1 = integerStringJavaPairRDD.sortByKey(false);
                JavaPairRDD<String, Integer> pairRDD = integerStringJavaPairRDD1.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
                        return new Tuple2<String, Integer>(integerStringTuple2._2, integerStringTuple2._1);
                    }
                });
                List<Tuple2<String, Integer>> take = pairRDD.take(3);
                for(Tuple2<String, Integer> t :take){
                    System.out.println(t._1 + ":" + t._2);
                }
                return stringIntegerJavaPairRDD;
            }
        });
        stringIntegerJavaPairDStream.print();

        jsc.start();
        try{
            jsc.awaitTermination();
        }catch (Exception e){
            e.printStackTrace();
        }

        jsc.close();

    }
}
